package com.cloud.mq.controller;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSONObject;
import com.cloud.common.core.domain.R;
import com.cloud.common.core.utils.JsonUtil;
import com.cloud.common.security.annotation.InnerAuth;
import com.cloud.mq.modo.MsgDto;
import com.cloud.mq.producer.KafkaProducer;
import com.cloud.system.api.domain.SysOperLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import java.util.Map;
import java.util.UUID;

@Slf4j
@RestController
@RequestMapping("kafakamq")
public class KafakaProducerController {

    @Autowired
    KafkaProducer kafkaProducer;

//生产者示例
    @InnerAuth
    @PostMapping("/saveLog")
    public R<Boolean> saveLog(@RequestBody SysOperLog sysOperLog){
        String json = JsonUtil.objToJson(sysOperLog);
        //序列化
        kafkaProducer.sendMessage("log_queue",json);
        return null;
    }

    private final String topicName = "hub-topic-city-delta";
    @GetMapping("/f01_1")
    public Object f01_1(String msgContent) {
        try {
            //3.获取业务数据对象
            String uuid= UUID.randomUUID().toString().replace("-","");
            long now=System.currentTimeMillis();
            String msgKey = "delta" + ":" + uuid + ":" + now;
            MsgDto msgDto = MsgDto.buildDto(uuid,now,msgContent);
            String msgData = JSONObject.toJSONString(msgDto);
            cn.hutool.json.JSONObject obj = JSONUtil.createObj();
            obj.set("msgKey",msgKey);
            obj.set("msgData",msgData);
            kafkaProducer.sendMessage(topicName,obj.toString());
        } catch (Exception e) {
            log.info("Producer写入Topic异常.");
//            持久化到数据库   定时扫描数据库是否存在应该生产但是没有发送到kafka的消息,kafka正常以后send消息
            e.printStackTrace();
        }
        return "写入成功";
    }

}